package com.mq.springboot.config;

import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.*;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.ConditionalRejectingErrorHandler;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.util.ErrorHandler;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * @author zhanghaidong
 * @version v1.0
 * @description
 * @date 2019/8/16
 */
@Configuration
public class RabbitMqConfig {

    private static final Logger logger = LoggerFactory.getLogger(RabbitMqConfig.class);

    public final static String QUEUE_NAME = "spring-boot-queue";
    public final static String EXCHANGE_NAME = "spring-boot-exchange";
    public final static String ROUTING_KEY = "spring-boot-key";

    /**
     * 创建队列
     * @return
     */
    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME, false, false, false, null);
    }

    /**
     * 创建一个 topic 类型的交换器
     * @return
     */
    @Bean
    public TopicExchange exchange() {

//        Exchange exchange = ExchangeBuilder.
//                topicExchange(EXCHANGE_NAME).
//                durable(false).
//                autoDelete().
//                withArguments(null).
//                build();
//        return (TopicExchange) exchange;

        return new TopicExchange(EXCHANGE_NAME, false, false, null);
    }

    /**
     * 使用路由键（routingKey）把队列（Queue）绑定到交换器（Exchange）
     * @param queue
     * @param exchange
     * @return
     */
    @Bean
    public Binding binding(Queue queue, TopicExchange exchange) {
        /*new Binding(queue.getName(),
                Binding.DestinationType.EXCHANGE,
                exchange.getName(),
                ROUTING_KEY,
                new HashMap<>(16));*/
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }

    /**
     * setAddresses：设置了rabbitmq的地址、端口，集群部署的情况下可填写多个，“,”分隔。
     * setUsername：设置rabbitmq的用户名。
     * setPassword：设置rabbitmq的用户密码。
     * setVirtualHost：设置virtualHost。
     * setCacheMode：设置缓存模式，共有两种，CHANNEL和CONNECTION模式。
     * 	    CHANNEL模式，程序运行期间ConnectionFactory会维护着一个Connection，所有的操作都会使用这个Connection，但一个Connection中可以有多个Channel，操作rabbitmq之前都必须先获取到一个Channel，否则就会阻塞（可以通过setChannelCheckoutTimeout()设置等待时间），这些Channel会被缓存（缓存的数量可以通过setChannelCacheSize()设置）；
     * 	    CONNECTION模式，这个模式下允许创建多个Connection，会缓存一定数量的Connection，每个Connection中同样会缓存一些Channel，除了可以有多个Connection，其它都跟CHANNEL模式一样。
     *
     * 	    这里的Connection和Channel是spring-amqp中的概念，并非rabbitmq中的概念
     *
     * setChannelCacheSize：设置每个Connection中（注意是每个Connection）可以缓存的Channel数量，注意只是缓存的Channel数量，不是Channel的数量上限，操作rabbitmq之前（send/receive message等）要先获取到一个Channel，获取Channel时会先从缓存中找闲置的Channel，如果没有则创建新的Channel，当Channel数量大于缓存数量时，多出来没法放进缓存的会被关闭。
     *       注意，改变这个值不会影响已经存在的Connection，只影响之后创建的Connection。
     * setChannelCheckoutTimeout：当这个值大于0时，channelCacheSize不仅是缓存数量，同时也会变成数量上限，从缓存获取不到可用的Channel时，不会创建新的Channel，会等待这个值设置的毫秒数，到时间仍然获取不到可用的Channel会抛出AmqpTimeoutException异常。
     *       同时，在CONNECTION模式，这个值也会影响获取Connection的等待时间，超时获取不到Connection也会抛出AmqpTimeoutException异常。
     *
     *       同时，在CONNECTION模式，这个值也会影响获取Connection的等待时间，超时获取不到Connection也会抛出AmqpTimeoutException异常。
     *
     * setPublisherReturns、setPublisherConfirms：producer端的消息确认机制（confirm和return），设为true后开启相应的机制，后文详述。
     *       官方文档描述publisherReturns设为true打开return机制，publisherComfirms设为true打开confirm机制，但测试结果（2.0.5.RELEASE版本）是，任意一个设为true，两个都会打开。
     *
     * addConnectionListener、addChannelListener、setRecoveryListener：添加或设置相应的Listener，后文详述。
     *
     * setConnectionCacheSize：仅在CONNECTION模式使用，设置Connection的缓存数量。
     *
     * setConnectionLimit：仅在CONNECTION模式使用，设置Connection的数量上限。
     *      上面的bean配置，除了需要注入的几个listener bean以外，其它设置的都是其默认值（2.0.5.RELEASE版本），后面的bean示例配置也是一样，部分属性不同版本的默认值可能有所不同。
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory(ChannelListener channelListener, ConnectionListener connectionListener, RecoveryListener recoveryListener) {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory("127.0.0.1", 5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
        connectionFactory.setChannelCacheSize(20);
        connectionFactory.setChannelCheckoutTimeout(1000L);
        connectionFactory.setPublisherReturns(true);
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setConnectionCacheSize(10);
        connectionFactory.setConnectionLimit(10);

        List<ChannelListener> channelListeners = new ArrayList<>();
        channelListeners.add(channelListener);
        connectionFactory.setChannelListeners(channelListeners);

        List<ConnectionListener> connectionListeners = new ArrayList<>();
        connectionListeners.add(connectionListener);
        connectionFactory.setConnectionListeners(connectionListeners);

        connectionFactory.setRecoveryListener(recoveryListener);
        return connectionFactory;
    }

    /**
     * setConnectionFactory：设置spring-amqp的ConnectionFactory。
     * setRetryTemplate：设置重试机制，详情见后文。
     * setMessageConverter：设置MessageConverter，用于java对象与Message对象（实际发送和接收的消息对象）之间的相互转换，详情见后文。
     * setChannelTransacted：打开或关闭Channel的事务，关于amqp的事务后文描述。
     *      amqp事务仅仅适用于publish和ack，rabbitmq增加了reject的事务。其它操作都不具备事务特性。也就是说，rabbitmq本身的事务可以保证producer端发出的消息成功被broker收到（不能保证一定会进入queue），consumer端发出的确认信息成功被broker收到，其它诸如consumer端具体的消费逻辑之类如果想要获得事务功能，需要引入外部事务。
     *      引入rabbitmq事务很简单，将RabbitTemplate或者RabbitListenerContainerFactory的channelTransacted属性设为true即可
     *
     * setReturnCallback、setConfirmCallback：return和confirm机制的回调接口，后文详述。
     * setMandatory：设为true使ReturnCallback生效。
     */
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        return rabbitTemplate;
    }

    /**
     *  配置RabbitListenerContainerFactory
     *      这个bean仅在consumer端通过@RabbitListener注解的方式接收消息时使用，每一个@RabbitListener注解的方法都会由这个RabbitListenerContainerFactory创建一个MessageListenerContainer，负责接收消息。
     *
     *  setConnectionFactory：设置spring-amqp的ConnectionFactory。
     *  setMessageConverter：对于consumer端，MessageConverter也可以在这里配置。
     *  setAcknowledgeMode：设置consumer端的应答模式，共有三种：NONE、AUTO、MANUAL。
     * 	    NONE，无应答，这种模式下rabbitmq默认consumer能正确处理所有发出的消息，所以不管消息有没有被consumer收到，有没有正确处理都不会恢复；
     * 	    AUTO，由Container自动应答，正确处理发出ack信息，处理失败发出nack信息，rabbitmq发出消息后将会等待consumer端的应答，只有收到ack确认信息才会把消息清除掉，收到nack信息的处理办法由setDefaultRequeueRejected()方法设置，所以在这种模式下，发生错误的消息是可以恢复的。
     * 	    MANUAL，基本同AUTO模式，区别是需要人为调用方法给应答。
     *  setConcurrentConsumers：设置每个MessageListenerContainer将会创建的Consumer的最小数量，默认是1个。
     *  setMaxConcurrentConsumers：设置每个MessageListenerContainer将会创建的Consumer的最大数量，默认等于最小数量。
     *  setPrefetchCount：设置每次请求发送给每个Consumer的消息数量。
     *  setChannelTransacted：设置Channel的事务。
     *  setTxSize：设置事务当中可以处理的消息数量。
     *  setDefaultRequeueRejected：设置当rabbitmq收到nack/reject确认信息时的处理方式，设为true，扔回queue头部，设为false，丢弃。
     *  setErrorHandler：实现ErrorHandler接口设置进去，所有未catch的异常都会由ErrorHandler处理。
     *
     *  @RabbitListener注解的方法中抛出的异常，首先会进入RabbitListenerErrorHandler，这里如果没有能力处理这个异常，需要将其重新抛出（否则不会进入ErrorHandler），然后异常将会进入ErrorHandler，一旦异常进入ErrorHandler就意味着消息消费失败了（所以不需要重新抛出异常）。
     *  RabbitListenerErrorHandler没有默认配置，ErrorHandler有一个默认的ConditionalRejectingErrorHandler类，它的处理方式是打印日志，然后辨别异常类型，
     */
    @Bean
    public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(CachingConnectionFactory cachingConnectionFactory,
                                                                               ErrorHandler errorHandler) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(cachingConnectionFactory);
        factory.setAcknowledgeMode(AcknowledgeMode.AUTO);
        factory.setConcurrentConsumers(1);
        factory.setMaxConcurrentConsumers(1);
        factory.setPrefetchCount(250);
        factory.setChannelTransacted(false);
        factory.setDefaultRequeueRejected(true);
        factory.setErrorHandler(errorHandler);
        return factory;
    }

    @Bean
    public ErrorHandler errorHandler() {
        return new ConditionalRejectingErrorHandler();
    }

    /**
     * ChannelListener接口，监听Channel的创建和异常关闭。
     * @return
     */
    @Bean
    public ChannelListener channelListener() {
        return new ChannelListener() {
            @Override
            public void onCreate(Channel channel, boolean transactional) {
                logger.info("channel number:{}, nextPublishSqlNo:{}",
                        channel.getChannelNumber(),
                        channel.getNextPublishSeqNo());
            }

            @Override
            public void onShutDown(ShutdownSignalException signal) {
                logger.error("channel shutdown, reason:{}, errorLevel:{}",
                        signal.getReason().protocolMethodName(),
                        signal.isHardError() ? "connection" : "channel");
            }
        };
    }

    /**
     * BlockedListener监听Connection的block和unblock。
     * @return
     */
    @Bean
    public BlockedListener blockedListener() {
        return new BlockedListener() {
            @Override
            public void handleBlocked(String reason) {
                logger.info("connection blocked, reason:{}", reason);
            }

            @Override
            public void handleUnblocked() {
                logger.info("connection unblocked");
            }
        };
    }

    /**
     * ConnectionListener监听Connection的创建、关闭和异常终止。
     * @return
     */
    @Bean
    public ConnectionListener connectionListener() {
        return new ConnectionListener() {
            @Override
            public void onCreate(Connection connection) {
                logger.info("connection created.");
            }

            @Override
            public void onClose(Connection connection) {
                logger.info("connection closed.");
            }

            @Override
            public void onShutDown(ShutdownSignalException signal) {
                logger.error("connection shutdown, reason:{}, errorLevel:{}",
                        signal.getReason().protocolMethodName(),
                        signal.isHardError() ? "connection" : "channel");
            }
        };
    }

    /**
     * RecoveryListener监听开始自动恢复Connection、自动恢复连接完成。
     * @return
     */
    @Bean
    public RecoveryListener recoveryListener() {
        return new RecoveryListener() {
            @Override
            public void handleRecovery(Recoverable recoverable) {
                logger.info("automatic recovery completed");
            }
            @Override
            public void handleRecoveryStarted(Recoverable recoverable) {
                logger.info("automatic recovery started");
            }
        };
    }
}
